package com.bytes32.preprocess;

import java.io.IOException;
import java.util.Random;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import com.bytes32.classification.tree.dataset.AttributeType;
import com.bytes32.classification.tree.dataset.Feature;
import com.bytes32.classification.tree.dataset.FeatureType;
import com.bytes32.classification.tree.dataset.InvalidFeatureValueException;
import com.bytes32.config.ParamsConfig;

/**
 * Mapper that samples the dataset and returns a feature index and feature value
 * + count Sampling without replacement
 * 
 * @author murariuf
 * 
 */
public class SamplingMapper implements Mapper<LongWritable, Text, IntWritable, Feature> {

	private int samplingRate = 1;

	private int pickOne = 0;

	private Random generator = new Random();
	
	private String separator = ","; /* CSV as default TODO: change that later */
	
	/**
	 * Id of the label
	 */
	private int labelId = -1;

	@Override
	public void configure(JobConf conf) {
		/* set the sampling rate */
		this.samplingRate = conf.getInt(ParamsConfig.SAMPLING_RATE.name(), 1);
		
		/* separator */
		this.separator = conf.get(ParamsConfig.SEPARATOR.name());
		
		/* get the label id, if not here then assume it's the last column in the dataset and assign -1 for now */
		this.labelId = conf.getInt(ParamsConfig.LABEL_ID.name(), -1);
		
		/* pick a subSample */
		this.pickOne = new Random().nextInt(this.samplingRate);
	}

	@Override
	public void close() throws IOException {

	}

	@Override
	public void map(LongWritable keyIn, Text valueIn, OutputCollector<IntWritable, Feature> out, Reporter rep)
			throws IOException {
		/* sampling me */
		if (samplingRate > 1) {
			int rand = generator.nextInt(samplingRate);
			if (rand != this.pickOne) 
				return;
		} 
		String line = valueIn.toString();
		String[] features = line.split(separator);
		for (int i = 0; i< features.length ; i++){
			/* for each feature */
			String featureValue = features[i];
			/* we change the id for the label to a negative value to make sure it arrives into the combiner first
			 * this allows us to determine the type of the label first and also to determine in which position it sits*/
			Feature feature = new Feature();
			feature.setFeatureType(FeatureType.UNKNOWN);
			feature.setAttributeType(AttributeType.UNKNOWN);
			try {
				feature.incrementCount(featureValue);
			} catch (InvalidFeatureValueException e) {
				throw new IOException("Unable to parse feature");
			}
			if ((labelId > 0 && this.labelId == i) || (labelId < 0 && i == features.length-1) ){
				out.collect(new IntWritable(-i), feature);
			} else {
				out.collect(new IntWritable(i), feature);
			}
		}
	}

}
